package com.geek.zookeeper.connect.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.geek.zookeeper.view.Config;
import com.geek.zookeeper.connect.Connect;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

@Slf4j
public class ConnectImpl extends Connect {

	private static final String PATH = "/config";

	private ZooKeeper zoo;
	private final ObjectMapper mapper = new ObjectMapper();

	public CountDownLatch countDownLatch = new CountDownLatch(1);

	@Override
	public ZooKeeper init() throws IOException, InterruptedException {

		log.info("链接中...");
		zoo = new ZooKeeper(CONNECT_URL, CONNECT_TIMEOUT, watchedEvent -> {
			log.info("链接成功");
			countDownLatch.countDown();
		});

		countDownLatch.await();

		return zoo;
	}

	@Override
	public void deleteData() throws InterruptedException, KeeperException {
		Stat state = new Stat();
		byte[] data = zoo.getData(PATH, false, state);
		int version = state.getVersion();
		log.info("当前删除的版本号:"+version);
		zoo.delete(PATH, version);
	}

	@Override
	public void createData() throws IOException, InterruptedException, KeeperException {
		Config config = new Config();
		config.setKey("test");
		config.setValue("valueTest");
		byte[] bytes = mapper.writeValueAsBytes(config);
		zoo.create(PATH, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	}

	@Override
	public void readData() throws IOException, InterruptedException, KeeperException {

		Watcher watcher = new Watcher() {
			@SneakyThrows
			@Override
			public void process(WatchedEvent watchedEvent) {
				byte[] data = zoo.getData(PATH, this, null);
				Config config = mapper.readValue(data, Config.class);
				log.info("读取到数据更新的结果:"+config);
			}
		};
		byte[] data = zoo.getData(PATH, watcher, null);
		Config config = mapper.readValue(data, Config.class);
		log.info("读取结果:"+config);

	}


}
